package com.zmj.tl.database.taos.highvolume;

import lombok.Getter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.HashMap;
import java.util.Map;

/**
 * @author zhangjp
 * @version 1.0
 * @title SQLWriter
 * @description
 * @create 2024/4/10 10:46
 */

public class SQLWriter {

    final static Logger logger = LoggerFactory.getLogger(SQLWriter.class);

    private Connection conn;
    private Statement stmt;

    /**
     * current number of buffered records
     */
    @Getter
    private int bufferedCount = 0;
    /**
     * Maximum number of buffered records.
     * Flush action will be triggered if bufferedCount reached this value,
     */
    private int maxBatchSize;


    /**
     * Maximum SQL length.
     */
    private int maxSQLLength = 800_000;

    /**
     * Map from table name to column values. For example:
     * "tb001" -> "(1648432611249,2.1,114,0.09) (1648432611250,2.2,135,0.2)"
     */
    private Map<String, String> tbValues = new HashMap<>();

    /**
     * Map from table name to tag values in the same order as creating stable.
     * Used for creating table.
     */
    private Map<String, String> tbTags = new HashMap<>();
    private Map<String, String> metricValues = new HashMap<>();
    private Map<String, String> subValues = new HashMap<>();

    public SQLWriter(int maxBatchSize) {
        this.maxBatchSize = maxBatchSize;
    }


    /**
     * Get Database Connection
     *
     * @return Connection
     * @throws SQLException SQLException
     */
    public static Connection getConnection() throws SQLException {
        //String jdbcURL = System.getenv("TDENGINE_JDBC_URL");
        String jdbcURL = "jdbc:TAOS-RS://192.168.130.73:6041/iot_prod?user=usr_app_szy&password=YXdkkxsvmv9C";
//        String jdbcURL = "jdbc:TAOS://192.168.130.73:6041/?user=usr_app_szy&password=YXdkkxsvmv9C";
//        String jdbcURL = "jdbc:TAOS-RS://192.168.130.53:6041?user=root&password=taosdata";
//        HikariConfig config = new HikariConfig();
//       // config.setDriverClassName(TSDBDriver.class.getName()); // 设置JDBC驱动
//        config.setJdbcUrl(jdbcURL);
//        // 设置连接池配置
//        config.setMaximumPoolSize(1000);
//        config.setConnectionTimeout(300000); // 连接超时时间3秒
//        config.setIdleTimeout(600000); // 空闲连接存活时间10分钟
//        config.setMaxLifetime(1800000); // 连接最大存活时间30分钟
//        //config.setLeakDetectionThreshold(15000); // 检测连接泄露的时间阈值5秒
//        config.setLeakDetectionThreshold(0); // 检测连接泄露的时间阈值5秒
//        HikariDataSource dataSource = new HikariDataSource(config);
//        return dataSource.getConnection();
        return DriverManager.getConnection(jdbcURL);
    }

    /**
     * Create Connection and Statement
     *
     * @throws SQLException SQLException
     */
    public void init() throws SQLException {
        conn = getConnection();
        stmt = conn.createStatement();
       // stmt.execute("use iot_prod");
    }

    /**
     * Convert raw data to SQL fragments, group them by table name and cache them in a HashMap.
     * Trigger writing when number of buffered records reached maxBachSize.
     *
     * @param line raw data get from task queue in format: tbName,ts,current,voltage,phase,location,groupId
     */
//    public void processLine(String line) throws SQLException {
//        bufferedCount += 1;
//        int firstComma = line.indexOf(',');
//        String tbName = line.substring(0, firstComma);
//        int lastComma = line.lastIndexOf(',');
//        int secondLastComma = line.lastIndexOf(',', lastComma - 1);
//        String value = "(" + line.substring(firstComma + 1, secondLastComma) + ") ";
//        if (tbValues.containsKey(tbName)) {
//            tbValues.put(tbName, tbValues.get(tbName) + value);
//        } else {
//            tbValues.put(tbName, value);
//        }
//        if (!tbTags.containsKey(tbName)) {
//            String location = line.substring(secondLastComma + 1, lastComma);
//            String groupId = line.substring(lastComma + 1);
//            String tagValues = "('" + location + "'," + groupId + ')';
//            tbTags.put(tbName, tagValues);
//        }
//        if (bufferedCount == maxBatchSize) {
//            flush();
//        }
//    }
    public void processLine(String line) throws SQLException {
        bufferedCount += 1;
        String[] split = line.split(",");
        String tbName = split[3];
        String sub = "(" + split[1] + ",'" + split[2] + "'," + split[12] + ")";
//        String sub = "(" + split[1] + ",'" + split[2] +  "')";
//        String tagValues = "('" + split[5] + "','" + split[6] + "','" + split[4] + "','" + split[7] + "','" +  split[8] + "','" +split[9] + "','" + split[10] + "','" + split[11] +"',"+ split[12] + ")";
        String tagValues = "('" + split[5] + "','" + split[6] + "','" + split[4] + "','" + split[7] + "','" + split[8] + "','" + split[9] + "','" + split[10] + "','" + split[11] + "')";
        StringBuilder sb = new StringBuilder("INSERT INTO ");
        sb.append(tbName).append(" USING ").append(split[0]).append(" TAGS ").append(tagValues).append(" VALUES ").append(sub);
        logger.info("入库sql语句:{}", sb.toString());
        executeSQL(sb.toString());
    }

    /**
     * Assemble INSERT statement using buffered SQL fragments in Map {@link SQLWriter#tbValues} and execute it.
     * In case of "Table does not exit" exception, create all tables in the sql and retry the sql.
     */
    public void flush() throws SQLException {
        StringBuilder sb = new StringBuilder("INSERT INTO ");
        for (Map.Entry<String, String> entry : tbValues.entrySet()) {
            String tableName = entry.getKey();
            String values = entry.getValue();
            String q = tableName + " values " + values + " ";
            if (sb.length() + q.length() > maxSQLLength) {
                executeSQL(sb.toString());
                logger.warn("increase maxSQLLength or decrease maxBatchSize to gain better performance");
                sb = new StringBuilder("INSERT INTO ");
            }
            sb.append(q);
        }
        executeSQL(sb.toString());
        tbValues.clear();
        bufferedCount = 0;
    }

    public void executeSQL(String sql) throws SQLException {
        try {
            stmt.executeUpdate(sql);
        } catch (SQLException e) {
            // convert to error code defined in taoserror.h
            int errorCode = e.getErrorCode() & 0xffff;
            if (errorCode == 0x2603) {
                // Table does not exist
                //createTables();
                executeSQL(sql);
            } else {
                logger.error("Execute SQL: {}", sql);
                throw e;
            }
        } catch (Throwable throwable) {
            logger.error("Execute SQL: {}", sql);
            throw throwable;
        }
//        finally {
//            try {
//                stmt.close();
//            } catch (SQLException e) {
//                logger.error("stmt 关闭失败",e);
//            }
//            try {
//                conn.close();
//            } catch (SQLException e) {
//                logger.error("conn 关闭失败",e);
//            }
//        }
    }

    /**
     * Create tables in batch using syntax:
     * <p>
     * CREATE TABLE [IF NOT EXISTS] tb_name1 USING stb_name TAGS (tag_value1, ...) [IF NOT EXISTS] tb_name2 USING stb_name TAGS (tag_value2, ...) ...;
     * </p>
     */
    private void createTables() throws SQLException {
        StringBuilder sb = new StringBuilder("CREATE TABLE ");
        for (String tbName : tbValues.keySet()) {
            String tagValues = tbTags.get(tbName);
            sb.append("IF NOT EXISTS ").append(tbName).append(" USING ").append(metricValues.get("tbValues")).append("  TAGS ").append(tagValues).append(" ");
        }
        String sql = sb.toString();
        try {
            stmt.executeUpdate(sql);
        } catch (Throwable throwable) {
            logger.error("Execute SQL: {}", sql);
            throw throwable;
        }
    }

    public boolean hasBufferedValues() {
        return bufferedCount > 0;
    }

    public void close() {
        try {
            stmt.close();
        } catch (SQLException e) {
            logger.error("stmt 关闭失败",e);
        }
        try {
            conn.close();
        } catch (SQLException e) {
            logger.error("conn 关闭失败",e);
        }
    }
}
